package com.tycmc.storm.topology;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.tycmc.common.util.Property;
import com.tycmc.system.AppContext;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;

public class TopologyLauncher {

    public static void main(String[] args) {
        try {
            AppContext.getSpringContext();
            // 解析拓扑结构字符串(JSON格式)
            TopologyBuilder builder;
            int workerNum;

            String topologyStructure = Property.getProperty("topologyconfig");

            // 动态配置
            JSONObject tsObj = JSONObject.parseObject(topologyStructure);
            // 拓扑节点配置数组
            JSONArray componentArray = tsObj.getJSONArray(TopologyConstant.COMPONENTS);
            // 利用JSON字串生成拓扑构建器
            builder = TopologyUtil.generateTopologyBuilder(componentArray);
            workerNum = tsObj.getIntValue("workerNum");// 拓扑工作进程数

            // 配置拓扑，自定义类型注册序列化
            Config conf = new Config();
            conf.setDebug(false);
            conf.setMaxSpoutPending(10000);
            //conf.setMessageTimeoutSecs(60000);


            if (args != null && args.length > 0) {
                conf.setNumWorkers(workerNum);
                StormSubmitter.submitTopology(args[0], conf, builder.createTopology());

            } else {
                LocalCluster cluster = new LocalCluster();
                cluster.submitTopology("stormTopology", conf, builder.createTopology());
                Utils.sleep(1000 * 60 * 600);// 运行600分钟后停止
                cluster.killTopology("stormTopology");
                cluster.shutdown();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
